package com.atguigu.kafka.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.TopicConfig;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * Copyright (C), 2022-2022, 姚兵
 * Author: 32210
 * Date: 2022/10/4 0:19
 * FileName: AdminAPI
 * Description:
 */
public class AdminAPI {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.25.23:9092");

        try (Admin admin = Admin.create(props)) {
            adminDetail(admin);
        }
    }

    public static void adminCreate(Admin admin) throws ExecutionException, InterruptedException {
        String topicName = "yao";
        int partitions = 3;
        short replicationFactor = 3;
        // Create a compacted topic
        CreateTopicsResult result = admin.createTopics(Collections.singleton(
                new NewTopic(topicName, partitions, replicationFactor)
                        .configs(Collections.singletonMap(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT))));

        // Call values() to get the result for a specific topic
        KafkaFuture<Void> future = result.values().get(topicName);

        // Call get() to block until the topic creation is complete or has failed
        // if creation failed the ExecutionException wraps the underlying cause.
        future.get();
    }


    public static void adminDetail(Admin admin) throws ExecutionException, InterruptedException {
        String topicName = "yao";
//        DescribeTopicsResult describeTopicsResult = admin.describeTopics(Collections.singleton(topicName));
        ListTopicsResult listTopicsResult = admin.listTopics();
        Collection<TopicListing> listings = listTopicsResult.listings().get();
        listings.stream().forEach(System.out::println);

        DescribeTopicsResult describeTopicsResult = admin.describeTopics(Collections.singleton(topicName));
        describeTopicsResult.values().forEach((k,v)->{
            System.out.println("key:"+k);
            System.out.println("value:"+v);
            try {
                TopicDescription description = v.get();
                System.out.println(description);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        });
//        Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments =new HashMap<>();
//        TopicPartition topicPartition = new TopicPartition(topicName, 0);
//        NewPartitionReassignment newPartitionReassignment=new NewPartitionReassignment(Arrays.asList(2,3));
//        reassignments.put(topicPartition, Optional.of(newPartitionReassignment));
//        AlterPartitionReassignmentsResult alterPartitionReassignmentsResult = admin.alterPartitionReassignments(reassignments);
//        KafkaFuture<Void> all = alterPartitionReassignmentsResult.all();
//        all.get();
    }
}
